package com.kqkj.dyrule.engine.functions;

import com.kqkj.dyrule.engine.util.ConfigNames;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class KafkaSourceBuilder {
    Config config;
    public KafkaSourceBuilder(){
        config = ConfigFactory.load();
    }
    public FlinkKafkaConsumer<String> build(String topic) {
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers",config.getString(ConfigNames.KAFKA_BOOTSTRAP_SERVERS));
        prop.setProperty("auto.offset.reset", config.getString(ConfigNames.KAFKA_OFFSET_AUTORESET));

        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), prop);
        return kafkaConsumer;


    }
}
